package com.opages.netty.demo.lesson.dark_horse2;

import com.opages.netty.demo.lesson.ByteBufferUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author daizhong.liu
 * @create 2021-12-30 14:52
 **/
@Slf4j
public class BossWorker {
    static class Boss {
        ServerSocketChannel serverSocketChannel;
        Selector selector;
        Worker[] workers;
        // 用于负载均衡的原子整数
        AtomicInteger robin = new AtomicInteger(0);
        public Boss(int port) {
            try {
                serverSocketChannel = ServerSocketChannel.open();
                serverSocketChannel.bind(new InetSocketAddress(port));
                serverSocketChannel.configureBlocking(false);
                selector = Selector.open();
                serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);

                // 创建固定数量的Worker
                workers = new Worker[4];
                for(int i = 0; i < workers.length; i++) {
                    workers[i] = new Worker("worker-"+i);
                }
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
        public void start() {
            try {
                while (true) {
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        // BossSelector负责Accept事件
                        if (key.isAcceptable()) {
                            log.debug("boss接收accept事件.");
                            // 建立连接
                            SocketChannel socket = serverSocketChannel.accept();
                            log.debug("connected...");
                            socket.configureBlocking(false);
                            // socket注册到Worker的Selector中
                            log.debug("before read...");
                            // 负载均衡，轮询分配Worker
                            workers[robin.getAndIncrement() % workers.length].register(socket);
                            log.debug("after read...");
                        }
                    }
                }
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    static class Worker implements Runnable {
        private Thread thread;
        private volatile Selector selector;
        private String name;
        private volatile boolean started = false;
        private ConcurrentLinkedQueue<Runnable> queue;

        public Worker(String name) {
            this.name = name;
        }
        /**
         * 流程如下：
         * 1、调用一个register，说明有一个accept过来，因此需要事先将Selector给阻塞住，因此第一次会先执行work本身的线程selector阻塞住，等待连接
         * 2、调用register说明有accept，也说明未来会有read事件，因此当register时就新建一个任务到队列中，还没注册read事件，而是将work本身的selector唤醒
         * 3、唤醒selector后，会调用队列中的read事件进行注册。
         * 4、注册成功后进行while循环就又会阻塞住，当有新的read事件过来就会进行处理。
         * @Author daizhong.liu
         * @Date 2021/12/31 10:48
         * @Param [socket]
         * @return
         **/
        public void register(final SocketChannel socket) throws IOException {
            // 只启动一次
            if (!started) {
                thread = new Thread(this, name);
                selector = Selector.open();
                queue = new ConcurrentLinkedQueue<>();
                thread.start();
                started = true;
            }

            // 向同步队列中添加SocketChannel的注册事件
            // 在Worker线程中执行注册事件
            queue.add(new Runnable() {
                @Override
                public void run() {
                    try {
                        log.debug(name+":worker队列中添加一个[读事件]任务.");
                        socket.register(selector, SelectionKey.OP_READ);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
            // 唤醒被阻塞的Selector
            // select类似LockSupport中的park，wakeup的原理类似LockSupport中的unpark
            log.debug(name+":worker唤醒selector...");
            selector.wakeup();
        }

        @Override
        public void run() {
            while (true) {
                try {
                    log.debug(name+":worker线程中selector阻塞.");
                    selector.select();
                    log.debug(name+":worker线程存在事件...");
                    // 通过同步队列获得任务并运行
                    Runnable task = queue.poll();
                    if (task != null) {
                        // 获得任务，执行注册操作
                        task.run();
                    }
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while(iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        // Worker只负责Read事件
                        if (key.isReadable()) {
                            // 简化处理，省略细节
                            SocketChannel socket = (SocketChannel) key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            socket.read(buffer);
                            buffer.flip();
                            ByteBufferUtil.debugAll(buffer);
                        }
                    }
                }catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        Boss boss = new Boss(8889);
        boss.start();
    }
}
